import os
import sys
import re

from datetime import datetime
from modules.base_upgrade import BaseUpgrade
from modules.compare_data import CompareData
from modules.install_db import InstallDb


class ManageDb(BaseUpgrade):
    def __init__(self, *args, **kwargs):
        super(ManageDb, self).__init__(**kwargs)
        self.hostnames = args
        self.prefixPort = kwargs['prefixPort']
        self.tpcc_path = kwargs['tpcc_path']
        self.ftp_vmap_path = kwargs['ftp_vmap_path']
        self.upgrade_user = ''
        self.direct_user = ''
        self.upgrade_port = ''
        self.direct_port = ''
        self.upgrade = dict()
        self.direct = dict()
        self.hosts = dict()
        self.ins = None
        self.up_tpcc_pname = 'upgrade_props.pg'
        self.di_tpcc_pname = 'direct_props.pg'
        self.cda = None
        self.rollback_times = 2
        self.diff_log_dir = os.path.join(os.getcwd(), 'diff')
        self.date = datetime.strftime(datetime.now(), '%m%d')
        self.prometheus_ip = '10.243.209.200'
        self.prometheus_user = 'opengauss_exporter'
        self.ip_lists = [self.ipaddress, self.ipaddress_std1,
                         self.ipaddress_std2]

    def init_data(self, up_user_prefix, di_user_prefix):
        self.upgrade_user = self.sh(f'ls /home|grep {up_user_prefix}')
        self.direct_user = self.sh(f'ls /home|grep {di_user_prefix}')
        cmd = f"grep -w 'port =' {self.db_root_path}/{self.upgrade_user}/" \
              f"cluster/dn1/postgresql.conf|awk '{{print $3}}'"
        self.upgrade_port = self.sh(cmd)
        self.direct_port = self.sh(
            f"grep -w 'port =' {self.db_root_path}/{self.direct_user}/"
            f"cluster/dn1/postgresql.conf|awk '{{print $3}}'")

        self.upgrade = {'ip': self.ipaddress,
                        'username': self.upgrade_user,
                        'pwd': self.passwd,
                        'db_name': 'postgres',
                        'db_port': self.upgrade_port,
                        'db_version': 'upgrade',
                        'db_env': f'/home/{self.upgrade_user}/gaussdb.bashrc'}
        self.direct = {'ip': self.ipaddress,
                       'username': self.direct_user,
                       'pwd': self.passwd,
                       'db_name': 'postgres',
                       'db_port': self.direct_port,
                       'db_version': 'direct',
                       'db_env': f'/home/{self.direct_user}/gaussdb.bashrc'}

        self.log.info(self.upgrade)
        self.log.info(self.direct)

    def install(self, user_prefix, db_version, ftp_path,
                cm_flag=False):
        try:
            self.log.info('安装数据库')
            in_params = dict()
            in_params['rootpasswd'] = self.passwd
            in_params['userpasswd'] = self.passwd
            in_params['dburpasswd'] = self.passwd
            in_params['func_user_prefix'] = user_prefix
            in_params['prefixPort'] = self.prefixPort
            in_params['db_root_path'] = self.db_root_path
            in_params['pkg_root_path'] = self.pkg_root_path
            in_params['db_version'] = db_version
            in_params['ftp_path'] = ftp_path

            self.ins = InstallDb(**in_params)
            self.hosts = dict(zip((self.ipaddress, self.ipaddress_std1,
                                   self.ipaddress_std2), self.hostnames))
            self.ins.clean_env(self.ipaddress, self.ipaddress_std1,
                               self.ipaddress_std2)
            self.ins.prepare_user()
            self.ins.get_pkg()
            self.ins.prepare_xml(**self.hosts)
            if cm_flag:
                self.ins.update_xml_for_cm(**self.hosts)
            self.ins.preinstall()
            self.ins.install(self.ipaddress_std2)
            self.ins.set_param(*self.hosts.keys())
            self.ins.get_db_vesion()
            if cm_flag:
                self.ins.check_status(True)
            else:
                self.ins.check_status()
        except Exception as e:
            print(str(e))
            raise Exception(f'安装{db_version}数据库失败')

    def get_dolphin_v(self, ssh, di=False):
        self.log.info('获取dolphin插件版本')
        run_sql = '\\dx'
        if di:
            res = self.ssh_run_sql(ssh, run_sql, 'db_testb', **self.direct)
            self.log.info(res)
            assert 'dolphin' in res
            pattern = r'\s*(\w+)\s*\|\s*(\d+\.\d+)\s*\|\s*(\w+)\s*\|\s*(.*)\s*'
            matches = re.findall(pattern, res)
            for match in matches:
                if match[0] == 'dolphin':
                    version = match[1]
                    self.log.info(f'dolphin版本：{version}')
                    return version

        res = self.ssh_run_sql(ssh, run_sql, 'db_testb', **self.upgrade)
        self.log.info(res)
        assert 'dolphin' in res
        pattern = r'\s*(\w+)\s*\|\s*(\d+\.\d+)\s*\|\s*(\w+)\s*\|\s*(.*)\s*'
        matches = re.findall(pattern, res)
        for match in matches:
            if match[0] == 'dolphin':
                version = match[1]
                self.log.info(f'dolphin版本：{version}')
                return version

    def compare_b_mode(self):
        ssh_up = self.ssh_connect(self.ipaddress, self.upgrade['username'],
                                  self.passwd)
        ssh_di = self.ssh_connect(self.ipaddress, self.direct['username'],
                                  self.passwd)
        self.log.info('先建B库')
        create_db_sql = "drop database if exists db_testb;" \
                        "create database db_testb dbcompatibility 'B';"
        res1 = self.ssh_run_sql(ssh_up, create_db_sql, **self.upgrade)
        res2 = self.ssh_run_sql(ssh_di, create_db_sql, **self.direct)
        self.log.info('升级后：' + res1)
        self.log.info(res2)
        assert 'CREATE DATABASE' in res1, res2

        self.log.info('再建表对比')
        run_sql = "select count(*) from pg_extension where extname='dolphin';"
        res1 = self.ssh_run_sql(ssh_up, run_sql, 'db_testb', **self.upgrade)
        res2 = self.ssh_run_sql(ssh_di, run_sql, 'db_testb', **self.direct)
        self.log.info(res1)
        self.log.info(res2)
        assert res1.splitlines()[-2].strip() == res2.splitlines()[
            -2].strip() == '1'

        run_sql = "create table tb_testb(id int,name text); desc tb_testb;"
        res1 = self.ssh_run_sql(ssh_up, run_sql, 'db_testb', **self.upgrade)
        res2 = self.ssh_run_sql(ssh_di, run_sql, 'db_testb', **self.direct)
        self.log.info(res1)
        self.log.info(res2)
        assert 'CREATE TABLE' in res1, res2
        assert '2 rows' in res1, res2

        self.log.info('再对比dolphin插件版本')
        res1 = self.get_dolphin_v(ssh_up)
        res2 = self.get_dolphin_v(ssh_di, di=True)
        assert res1 == res2

        self.ssh_close(ssh_up)
        self.ssh_close(ssh_di)

    def compare_sys(self):
        ssh_up = self.ssh_connect(self.ipaddress, self.upgrade['username'],
                                  self.passwd)
        ssh_di = self.ssh_connect(self.ipaddress, self.direct['username'],
                                  self.passwd)
        self.log.info('系统表校验，先检验表名')
        diff_log_path = os.path.join(self.diff_log_dir, 'pg_tables.html')
        pg_tables_cmd = 'select distinct (quote_literal(schemaname),' \
                        'quote_literal(tablename)) as a ' \
                        'from pg_tables order by a;'
        text1 = self.ssh_run_sql(ssh_up, pg_tables_cmd, **self.upgrade)
        text2 = self.ssh_run_sql(ssh_di, pg_tables_cmd, **self.direct)
        self.compare_text(pg_tables_cmd, text1, text2, diff_log_path)

        self.log.info('系统表校验，再校验表结构')
        t_list = [list(eval(i)) for i in text1.splitlines() if
                  i.strip().startswith("('")]
        for one in t_list:
            t_cmd = f'\\d+ {one[0]}.{one[1]}'
            t1 = self.ssh_run_sql(ssh_up, t_cmd, **self.upgrade)
            t2 = self.ssh_run_sql(ssh_di, t_cmd, **self.direct)
            self.compare_text(t_cmd, t1, t2, diff_log_path)

        self.log.info('系统视图校验，先检验视图名')
        diff_log_path = os.path.join(self.diff_log_dir, 'pg_views.html')
        pg_views_cmd = 'select distinct (quote_literal(schemaname),' \
                       'quote_literal(viewname)) as a ' \
                       'from pg_views order by a;'
        text1 = self.ssh_run_sql(ssh_up, pg_views_cmd, **self.upgrade)
        text2 = self.ssh_run_sql(ssh_di, pg_views_cmd, **self.direct)
        self.compare_text(pg_views_cmd, text1, text2, diff_log_path)

        self.log.info('系统视图校验，再校验视图结构')
        v_list = [list(eval(i)) for i in text1.splitlines() if
                  i.strip().startswith("('")]
        for one in v_list:
            v_cmd = f'\\d+ {one[0]}.{one[1]}'
            v1 = self.ssh_run_sql(ssh_up, v_cmd, **self.upgrade)
            v2 = self.ssh_run_sql(ssh_di, v_cmd, **self.direct)
            self.compare_text(v_cmd, v1, v2, diff_log_path)

        self.ssh_close(ssh_up)
        self.ssh_close(ssh_di)

    def compare_meta(self):
        ssh_up = self.ssh_connect(self.ipaddress, self.upgrade['username'],
                                  self.passwd)
        ssh_di = self.ssh_connect(self.ipaddress, self.direct['username'],
                                  self.passwd)
        diff_log_path = os.path.join(self.diff_log_dir, 'meta.html')

        self.log.info('开始执行元命令')
        meta_cmd = '\\?'
        text1 = self.ssh_run_sql(ssh_up, meta_cmd, **self.upgrade)
        text2 = self.ssh_run_sql(ssh_di, meta_cmd, **self.direct)
        self.compare_text(meta_cmd, text1, text2, diff_log_path)

        meta_cmd = '\\copyright'
        text1 = self.ssh_run_sql(ssh_up, meta_cmd, **self.upgrade)
        text2 = self.ssh_run_sql(ssh_di, meta_cmd, **self.direct)
        self.compare_text(meta_cmd, text1, text2, diff_log_path)

        self.log.info('因\g指定文件路径后，只能在同一session中生效，所以无法验证查询写入的情况')
        f_path = f'~/test_g'
        clear_cmd = f'> {f_path}'
        self.ssh_run(ssh_up, clear_cmd)
        self.ssh_run(ssh_di, clear_cmd)
        meta_cmd = f'\\g {f_path}'
        text1 = self.ssh_run_sql(ssh_up, meta_cmd, **self.upgrade)
        text2 = self.ssh_run_sql(ssh_di, meta_cmd, **self.direct)
        self.compare_text(meta_cmd, text1, text2, diff_log_path)
        clean_cmd = f'rm -rf {f_path}'
        self.ssh_run(ssh_up, clean_cmd)
        self.ssh_run(ssh_di, clean_cmd)

        for c in ('\\h', '\\help', '\\h *', '\\help *'):
            text1 = self.ssh_run_sql(ssh_up, c, **self.upgrade)
            text2 = self.ssh_run_sql(ssh_di, c, **self.direct)
            self.compare_text(c, text1, text2, diff_log_path)

        for c in ('\\parallel', '\\parallel on', '\\parallel on 2',
                  '\\parallel off'):
            text1 = self.ssh_run_sql(ssh_up, c, **self.upgrade)
            text2 = self.ssh_run_sql(ssh_di, c, **self.direct)
            self.compare_text(c, text1, text2, diff_log_path)

        meta_cmd = '\\q'
        text1 = self.ssh_run_sql(ssh_up, meta_cmd, **self.upgrade)
        text2 = self.ssh_run_sql(ssh_di, meta_cmd, **self.direct)
        self.compare_text(meta_cmd, text1, text2, diff_log_path)

        f_path = '~/test_e'
        self.ssh_run(ssh_up, f'echo "111111"> {f_path}')
        self.ssh_run(ssh_di, f'echo "111111"> {f_path}')
        meta_cmd = f'\\e {f_path}'
        text1 = self.ssh_run_sql(ssh_up, meta_cmd, **self.upgrade)
        text2 = self.ssh_run_sql(ssh_di, meta_cmd, **self.direct)
        self.compare_text(meta_cmd, text1, text2, diff_log_path)
        self.ssh_run(ssh_up, f'rm -rf {f_path}')
        self.ssh_run(ssh_di, f'rm -rf {f_path}')

        for c in ('\\ef ', '\\p ', '\\r '):
            text1 = self.ssh_run_sql(ssh_up, c, **self.upgrade)
            text2 = self.ssh_run_sql(ssh_di, c, **self.direct)
            self.compare_text(c, text1, text2, diff_log_path)

        create_func = '''create function f_test(num1 integer, num2 integer) \
            return integer
            as
            begin
            return num1 + num2;
            end;'''
        up_res = self.ssh_run_sql(ssh_up, create_func, **self.upgrade)
        di_res = self.ssh_run_sql(ssh_di, create_func, **self.direct)
        print(f'up_res:{up_res}    di_res:{di_res}')
        for c in ('\\sf f_test', '\\sf+ f_test',
                  '\\ef f_test'):
            text1 = self.ssh_run_sql(ssh_up, c, **self.upgrade)
            text2 = self.ssh_run_sql(ssh_di, c, **self.direct)
            self.compare_text(c, text1, text2, diff_log_path)
        drop_func = f'drop function f_test;'
        self.ssh_run_sql(ssh_up, drop_func, **self.upgrade)
        self.ssh_run_sql(ssh_di, drop_func, **self.direct)

        f_path = f'~/test_w'
        self.ssh_run(ssh_up, f'mkdir {f_path}')
        self.ssh_run(ssh_di, f'mkdir {f_path}')
        meta_cmd = f'\\w {f_path}'
        text1 = self.ssh_run_sql(ssh_up, meta_cmd, **self.upgrade)
        text2 = self.ssh_run_sql(ssh_di, meta_cmd, **self.direct)
        self.compare_text(meta_cmd, text1, text2, diff_log_path)
        self.ssh_run(ssh_up, f'rm -rf {f_path}')
        self.ssh_run(ssh_di, f'rm -rf {f_path}')

        create_table = f"drop table if exists t_test_copy;" \
                       f"create table t_test_copy(a int, b varchar2(10));" \
                       f"insert into t_test_copy values(1,'aa'),(2,'bb');"
        self.ssh_run_sql(ssh_up, create_table, **self.upgrade)
        self.ssh_run_sql(ssh_di, create_table, **self.direct)
        f_path = '/home/test_copy_table.csv'
        meta_cmd = f'\\copy t_test_copy to {f_path}'
        text1 = self.ssh_run_sql(ssh_up, meta_cmd, **self.upgrade)
        text2 = self.ssh_run_sql(ssh_di, meta_cmd, **self.direct)
        self.compare_text(meta_cmd, text1, text2, diff_log_path)
        drop_table = f'drop table t_test_copy;'
        self.ssh_run_sql(ssh_up, drop_table, **self.upgrade)
        self.ssh_run_sql(ssh_di, drop_table, **self.direct)
        rm_cmd = f'rm -rf {f_path}'
        self.ssh_run(ssh_up, rm_cmd)
        self.ssh_run(ssh_di, rm_cmd)

        f_path = '~/test_file'
        self.ssh_run(ssh_up, f'echo "select * from pg_user;" > {f_path}')
        self.ssh_run(ssh_di, f'echo "select * from pg_user;" > {f_path}')
        for c in ('\\echo "testing"', '\\qecho test ok', f'\\i {f_path}',
                  f'\\ir {f_path}', f'\\o {f_path}'):
            text1 = self.ssh_run_sql(ssh_up, c, **self.upgrade)
            text2 = self.ssh_run_sql(ssh_di, c, **self.direct)
            self.compare_text(c, text1, text2, diff_log_path)
        rm_cmd = f'rm -rf {f_path}'
        self.ssh_run(ssh_up, rm_cmd)
        self.ssh_run(ssh_di, rm_cmd)

        self.log.info('使用gs_dump工具导出一个加密文件')
        f_path = '~/expect_file'
        create_table = f"drop table if exists table_test;" \
                       f"create table table_test(a int, b varchar2(10));"
        self.ssh_run_sql(ssh_up, create_table, **self.upgrade)
        self.ssh_run_sql(ssh_di, create_table, **self.direct)
        init_cmd = f'source {self.upgrade["db_env"]};' \
                   f'gs_dump -p {self.upgrade["db_port"]} postgres -f {f_path} ' \
                   f'-t table_test --with-encryption=AES128 --with-key=testing@123'
        self.ssh_run(ssh_up, init_cmd)
        init_cmd = f'source {self.direct["db_env"]};' \
                   f'gs_dump -p {self.direct["db_port"]} postgres -f {f_path} ' \
                   f'-t table_test --with-encryption=AES128 --with-key=testing@123'
        self.ssh_run(ssh_di, init_cmd)
        for c in (f'\\i+ {f_path}', f'\\ir+ {f_path}'):
            text1 = self.ssh_run_sql(ssh_up, c, **self.upgrade)
            text2 = self.ssh_run_sql(ssh_di, c, **self.direct)
            self.compare_text(c, text1, text2, diff_log_path)
        drop_table = f"drop table table_test;"
        self.ssh_run_sql(ssh_up, drop_table, **self.upgrade)
        self.ssh_run_sql(ssh_di, drop_table, **self.direct)
        rm_cmd = f'rm -rf {f_path}'
        self.ssh_run(ssh_up, rm_cmd)
        self.ssh_run(ssh_di, rm_cmd)

        self.log.info('显示信息元命令')
        create_table = f"drop table if exists t_test_d;" \
                       f"create table t_test_d(a int, b varchar2(10));"
        self.ssh_run_sql(ssh_up, create_table, **self.upgrade)
        self.ssh_run_sql(ssh_di, create_table, **self.direct)
        for c in ('\\d', '\\dS+ ', '\\d t_test_d', '\\dS+ t_test_d',
                  '\\d+', '\\dS ', '\\d+ t_test_d', '\\dS t_test_d'):
            text1 = self.ssh_run_sql(ssh_up, c, **self.upgrade)
            text2 = self.ssh_run_sql(ssh_di, c, **self.direct)
            self.compare_text(c, text1, text2, diff_log_path)
        drop_table = f'drop table t_test_d;'
        self.ssh_run_sql(ssh_up, drop_table, **self.upgrade)
        self.ssh_run_sql(ssh_di, drop_table, **self.direct)

        for c in ('\\da', '\\daS', '\\da f*', '\\daS f*',
                  '\\db', '\\db+', '\\db p*', '\\db+ p*',
                  '\\dc', '\\dcS', '\\dc+', '\\dcS+',
                  '\\dc a*', '\\dc+ a*', '\\dcS a*', '\\dcS+ a*',
                  '\\dC', '\\dC+', '\\dC o*', '\\dC+ o*',
                  '\\dd', '\\ddS', '\\ddp', '\\dD', '\\dDS', '\\dD+',
                  '\\dDS+', '\\ded', '\\ded+', '\\det', '\\det+',
                  '\\des', '\\des+', '\\des g*', '\\des g*', '\\deu',
                  '\\deu+', '\\dew', '\\dew+', '\\dew d*', '\\dew+ d*'):
            text1 = self.ssh_run_sql(ssh_up, c, **self.upgrade)
            text2 = self.ssh_run_sql(ssh_di, c, **self.direct)
            self.compare_text(c, text1, text2, diff_log_path)

        for c in ('\\df', '\\df antf', '\\df antfS+',
                  '\\df a*', '\\dfS a*', '\\dfS+ a*', '\\df antf a*',
                  '\\df antfS+ a*', '\\dF', '\\dF+', '\\dF s*', '\\dF+ s*',
                  '\\dFd', '\\dFd+', '\\dFd s*', '\\dFd+ s*',
                  '\\dFp', '\\dFp+', '\\dFp s*', '\\dFp+ s*',
                  '\\dFp', '\\dFp+', '\\dFp n*', '\\dFp+ n*',
                  '\\dg', '\\dg+', '\\dg t*', '\\dg+ t*',
                  '\\di', '\\di+', '\\diS+', '\\diS', '\\diS+ s*', '\\diS s*'):
            text1 = self.ssh_run_sql(ssh_up, c, **self.upgrade)
            text2 = self.ssh_run_sql(ssh_di, c, **self.direct)
            self.compare_text(c, text1, text2, diff_log_path)

        for c in ('\\dl', '\\dL ', '\\dL+', '\\dLS', '\\dLS+',
                  '\\dL j*', '\\dL+ j*', '\\dLS j*', '\\dLS+ j*',
                  '\\dm ', '\\dm+', '\\dmS', '\\dmS+',
                  '\\dm j*', '\\dm+ j*', '\\dmS j*', '\\dmS+ j*',
                  '\\dn ', '\\dn+', '\\dnS', '\\dnS+',
                  '\\dn d*', '\\dn+ d*', '\\dnS d*', '\\dnS+ d*',
                  '\\do ', '\\doS', '\\do #', '\\doS #',
                  '\\dO', '\\dO+', '\\dOS', '\\dOS+',
                  '\\dO d*', '\\dO+ d*', '\\dOS d*', '\\dOS+ d*',):
            text1 = self.ssh_run_sql(ssh_up, c, **self.upgrade)
            text2 = self.ssh_run_sql(ssh_di, c, **self.direct)
            self.compare_text(c, text1, text2, diff_log_path)

        for c in ('\\dp+', '\\dp t*', '\\drds', '\\drds * postgres',
                  '\\ds', '\\ds+', '\\dsS', '\\dsS+',
                  '\\dt', '\\dt+', '\\dtS', '\\dtS+', '\\dt pg*',
                  '\\dt t*', '\\dt+ t*', '\\dtS t*', '\\dtS+ t*',
                  '\\dT', '\\dT+', '\\dTS', '\\dTS+',
                  '\\dT b*', '\\dT+ b*', '\\dTS b*', '\\dTS+ b*',
                  '\\du', '\\du+', '\\du u*', '\\du+ u*',
                  '\\dv', '\\dv+', '\\dvS', '\\dvS+', '\\dv pg*',
                  '\\dv p*', '\\dv+ p*', '\\dvS p*', '\\dvS+ p*',
                  '\\dE', '\\dE+', '\\dES', '\\dES+',
                  '\\dx', '\\dx+', '\\dx d*', '\\dx+ d*',
                  '\\l', '\\l+', '\\z', '\\z t*'):
            text1 = self.ssh_run_sql(ssh_up, c, **self.upgrade)
            text2 = self.ssh_run_sql(ssh_di, c, **self.direct)
            self.compare_text(c, text1, text2, diff_log_path)

        self.log.info('格式化元命令')
        for c in ('\\a', '\\C', '\\C test_title', '\\f',
                  '\\f field_separator', '\\f |', '\\H'):
            text1 = self.ssh_run_sql(ssh_up, c, **self.upgrade)
            text2 = self.ssh_run_sql(ssh_di, c, **self.direct)
            self.compare_text(c, text1, text2, diff_log_path)

        for c in ('\\pset border 1', '\\pset expanded', '\\pset fieldsep',
                  '\\pset fieldsep_zero', '\\pset footer', '\\pset format u',
                  '\\pset format a', '\\pset null', '\\pset numericlocale',
                  '\\pset numericlocale on', '\\pset numericlocale off',
                  '\\pset pager', '\\pset pager on', '\\pset pager always',
                  '\\pset pager off', '\\pset recordsep',
                  '\\pset recordsep_zero',
                  '\\pset tableattr', '\\pset title', '\\pset tuples_only',
                  '\\t', '\\t on', '\\t off', '\\T',
                  '\\x', '\\x on', '\\x off', '\\x auto',):
            text1 = self.ssh_run_sql(ssh_up, c, **self.upgrade)
            text2 = self.ssh_run_sql(ssh_di, c, **self.direct)
            self.compare_text(c, text1, text2, diff_log_path)

        self.log.info('连接元命令')
        for c in ('\\c template0', '\\c postgres',
                  '\\encoding', '\\encoding UTF8', '\\conninfo'):
            text1 = self.ssh_run_sql(ssh_up, c, **self.upgrade)
            text2 = self.ssh_run_sql(ssh_di, c, **self.direct)
            self.compare_text(c, text1, text2, diff_log_path)

        self.log.info('操作系统元命令')
        for c in ('\\cd /home', '\\setenv JAVA', '\\timing ',
                  '\\timing on', '\\timing off', '\\! ls -ld /home'):
            text1 = self.ssh_run_sql(ssh_up, c, **self.upgrade)
            text2 = self.ssh_run_sql(ssh_di, c, **self.direct)
            self.compare_text(c, text1, text2, diff_log_path)

        self.log.info('操作系统元命令')
        for c in ('\\set a 1', '\\set b', '\\unset a',
                  '\\set verbosity default', '\\set verbosity verbose',
                  '\\set verbosity terse', '\\set on_error_stop on',
                  '\\set on_error_stop off', '\\set retry 6'):
            text1 = self.ssh_run_sql(ssh_up, c, **self.upgrade)
            text2 = self.ssh_run_sql(ssh_di, c, **self.direct)
            self.compare_text(c, text1, text2, diff_log_path)

        self.log.info('大对象元命令')
        f_path = '~/test_file'
        init_cmd = f'echo "select * from pg_user;" > {f_path}'
        self.ssh_run(ssh_up, init_cmd)
        self.ssh_run(ssh_di, init_cmd)
        for c in (f'\\lo_import {f_path}', f'\\lo_export loboid {f_path}',
                  '\\lo_list', '\\lo_unlink loboid'):
            text1 = self.ssh_run_sql(ssh_up, c, **self.upgrade)
            text2 = self.ssh_run_sql(ssh_di, c, **self.direct)
            self.compare_text(c, text1, text2, diff_log_path)
        rm_cmd = f'rm -rf {f_path}'
        self.ssh_run(ssh_up, rm_cmd)
        self.ssh_run(ssh_di, rm_cmd)

        self.log.info('执行元命令完成')
        self.ssh_close(ssh_up)
        self.ssh_close(ssh_di)

    def compare_permissions(self):
        ssh_up = self.ssh_connect(self.ipaddress, self.upgrade['username'],
                                  self.passwd)
        ssh_di = self.ssh_connect(self.ipaddress, self.direct['username'],
                                  self.passwd)
        diff_log_path = os.path.join(self.diff_log_dir, 'permissions.html')

        self.log.info('元命令查看权限(模式、库、表、视图、序列)')
        for c in ('\\dn+', '\\l+', '\\dp+'):
            text1 = self.ssh_run_sql(ssh_up, c, **self.upgrade)
            text2 = self.ssh_run_sql(ssh_di, c, **self.direct)
            self.compare_text(c, text1, text2, diff_log_path)

        self.log.info('常用系统表和系统视图校验一致性（内含权限字段）')
        for c in ('\\dt pg*', '\\dv pg*'):
            text1 = self.ssh_run_sql(ssh_up, c, **self.upgrade)
            text2 = self.ssh_run_sql(ssh_di, c, **self.direct)
            self.compare_text(c, text1, text2, diff_log_path)

        self.log.info('权限校验，先查询所有角色')
        get_role_cmd = 'select rolname from pg_roles order by rolname;'
        rolnames = self.ssh_run_sql(ssh_up, get_role_cmd, **self.upgrade)
        per_cmd = list()
        for role in rolnames.splitlines()[2:-1]:
            self.log.info(f'role: {role}')
            self.log.info('权限校验，再通过角色查询所拥有权限')
            per_cmd.append(
                f"select * from pg_authid where rolname='{role.strip()}';")
            per_cmd.append(
                f"select * from pg_roles where rolname='{role.strip()}';")
            per_cmd.append(
                f"select * from information_schema.table_privileges "
                f"where grantee='{role.strip()}' order by 3, 4, 5, 6, 7;")
            per_cmd.append(
                f"select * from information_schema.usage_privileges "
                f"where grantee='{role.strip()}' order by 3, 4, 5;")
            per_cmd.append(
                f"select * from information_schema.routine_privileges "
                f"where grantee='{role.strip()}' order by 3, 4, 5;")
            per_cmd.append(
                f"select * from information_schema.column_privileges "
                f"where grantee='{role.strip()}'order by 3, 4, 5, 6;")
            per_cmd.append(
                f"select * from information_schema.udt_privileges "
                f"where grantee='{role.strip()}'3, 4, 5;")
            for per_c in per_cmd:
                text1 = self.ssh_run_sql(ssh_up, per_c, **self.upgrade)
                if role.strip() == self.upgrade_user:
                    per_c = per_c.replace(self.upgrade_user, self.direct_user)
                text2 = self.ssh_run_sql(ssh_di, per_c, **self.direct)
                self.compare_text(per_c, text1, text2, diff_log_path)
        self.log.info('权限对比完成')

        self.ssh_close(ssh_up)
        self.ssh_close(ssh_di)

    def init_compare_data(self, init_user, db_user, db_pwd, db_name, db_port):
        db_info = {'init_user': init_user,
                   'db_user': db_user,
                   'password': db_pwd,
                   'database': db_name,
                   'port': db_port,
                   'env_path': os.path.join('/', 'home', init_user,
                                            'gaussdb.bashrc')}
        self.cda = CompareData(**db_info)

    def get_table_data(self, lable):
        self.log.info('读取表数据')
        self.cda.get_tables()
        self.cda.get_checksum()
        self.cda.write_checksum_to_file(lable)
        self.log.info('读取表数据完成')

    def set_tpcc_props(self, db_host, db_port, db_name, db_user, db_pwd,
                       p_name='props.pg'):
        self.log.info('配置benchmark配置文件')
        sed_cmd = f"cd {self.tpcc_path} && " \
                  f"sed -i '3cconn=jdbc:postgresql://{db_host}:{db_port}/" \
                  f"{db_name}' {p_name} && " \
                  f"sed -i '4cuser={db_user}' {p_name} && " \
                  f"sed -i '5cpassword={db_pwd}' {p_name}"
        try:
            self.sh(sed_cmd, ispwd=True)
        except Exception as e:
            self.log.info(f'配置benchmark配置文件失败，ERROR: {str(e)}')
        else:
            self.log.info('配置benchmark配置文件完成')

    def run_tpcc(self, mode='run', p_name='props.pg'):
        self.log.info(f'开始执行benchmark-{mode}')
        if mode == 'run':
            s_name = 'runBenchmark.sh'
        elif mode == 'build':
            s_name = 'runDatabaseBuild.sh'
        else:
            s_name = 'runDatabaseDestroy.sh'
        run_tpcc_cmd = f"cd {self.tpcc_path} && sh {s_name} {p_name};"
        try:
            self.sh(run_tpcc_cmd)
        except Exception as e:
            self.log.info(f'执行benchmark-{mode}失败，ERROR: {str(e)}')
        else:
            self.log.info(f'执行benchmark-{mode}成功')

    def loading_gs_upgradechk(self, phase='after_upgrade'):
        """
        不支持元数据校验工具的数据库，加载元数据校验工具
        :param phase: 区分升级前、后、直接安装数据库的安装包路径
        :return: 无
        """
        if phase == 'before_upgrade':
            pkg = os.path.join(self.pkg_root_path, self.upgrade_user,
                               self.date, 'pkg')
        elif phase == 'after_upgrade':
            pkg = os.path.join(self.pkg_root_path, self.upgrade_user,
                               self.date, 'upgrade_pkg')
        elif phase == 'direct_install':
            pkg = os.path.join(self.pkg_root_path, self.direct_user,
                               self.date, 'pkg')
        with self.ssh_connect(self.ipaddress, 'root', self.passwd) as ssh:
            cmd = f'cd {pkg} && ' \
                  f'wget {self.ftp_vmap_path} && ' \
                  f'tar -zxvf "{self.ftp_vmap_path.split("/")[-1]}"'
            self.ssh_run(ssh, cmd, False, False)
            cmd = f"cd {pkg}/script && " \
                  f"sed -i '133,134 s/^/#/' upgrade_checker/opengauss.py && " \
                  f"chmod -R 755 {pkg} && " \
                  f"chmod -R 755 {pkg}/script/upgrade_checker/opengauss.py && " \
                  f"chmod -R 755 {pkg}/script/upgrade_checker/"
            self.ssh_run(ssh, cmd, False, True)

    def gs_upgradechk_export(self, phase='after_upgrade', **kwargs):
        """
        使用gs_upgradechk工具导出vmap。在屏幕打印信息中获取vmap位置
        :param phase: 区分升级前、后、直接安装数据库的安装包路径
        :param kwargs: 环境信息
        :return: 返回vmap路径
        """
        if phase == 'before_upgrade':
            pkg = os.path.join(self.pkg_root_path, self.upgrade_user,
                               self.date, 'pkg')
        elif phase == 'direct_install':
            pkg = os.path.join(self.pkg_root_path, self.direct_user,
                               self.date, 'pkg')

        with self.ssh_connect(self.ipaddress, kwargs['username'],
                              self.passwd) as ssh:
            output = ''
            cmd = f"source {kwargs['db_env']}; " \
                  f"gs_upgradechk export -p {kwargs['db_port']}"
            try:
                output = self.ssh_run(ssh, cmd, False, True)
            except Exception:
                pass

            if 'gs_upgradechk: command not found' in output:
                source = f'{pkg}/script/gs_upgradechk'
                self.log.info(source)
                get_gausshome = f"source {kwargs['db_env']}; " \
                                f"echo $GAUSSHOME"
                gausshome = self.ssh_run(ssh, get_gausshome, False, True)
                target = os.path.join(gausshome, 'bin', 'gs_upgradechk')
                self.log.info(target)
                ln_cmd = f'ln -s {source} {target}'
                self.log.info(ln_cmd)
                self.ssh_run(ssh, ln_cmd, False, True)
                output = self.ssh_run(ssh, cmd, False, False)
        res = output.splitlines()[-1].split(':')[-1]

        with self.ssh_connect(self.ipaddress, 'root', self.passwd) as ssh:
            if phase == 'before_upgrade':
                self.ssh_run(ssh, f'cd {self.db_root_path} && '
                                  f'rm -rf EXPORT-*')
            self.ssh_run(ssh, f'mv -i {res} {self.db_root_path}')
            vamp_path = os.path.join(self.db_root_path,
                                     res.split('/')[-1].strip()).replace('\\',
                                                                         '/')
            self.ssh_run(ssh, f'chmod -R 777 {vamp_path}')
            self.log.info(f'获取到的vmap:{vamp_path}')
            return vamp_path

    def gs_upgradechk_verify(self, vamp=None, phase='after_upgrade', **kwargs):
        """
        使用gs_upgradechk工具校验元数据。在屏幕打印信息中获取报告地址，读取报告检测相关字段
        :param vamp: 元数据校验地图，使用gs_upgradechk_export获取
        :param phase: 区分升级前、后、直接安装数据库的安装包路径
        :param kwargs: 环境信息
        :return: 校验结果、报告地址
        """
        if phase == 'before_upgrade' or phase == 'after_rollback':
            pkg = os.path.join(self.pkg_root_path, self.upgrade_user,
                               self.date, 'pkg')
        elif phase == 'after_upgrade':
            pkg = os.path.join(self.pkg_root_path, self.upgrade_user,
                               self.date, 'upgrade_pkg')

        cmd = f"source {kwargs['db_env']}; " \
              f"gs_upgradechk verify -p {kwargs['db_port']} -v {vamp}"
        with self.ssh_connect(self.ipaddress, kwargs['username'],
                              self.passwd) as ssh:
            output = ''
            try:
                output = self.ssh_run(ssh, cmd, False, True)
            except Exception:
                pass

            if 'gs_upgradechk: command not found' in output:
                source = f'{pkg}/script/gs_upgradechk'
                self.log.info(source)
                get_gausshome = f"source {kwargs['db_env']}; " \
                                f"echo $GAUSSHOME"
                gausshome = self.ssh_run(ssh, get_gausshome, False, True)
                target = os.path.join(gausshome, 'bin', 'gs_upgradechk')
                self.log.info(target)
                ln_cmd = f'ln -s {source} {target}'
                self.log.info(ln_cmd)
                self.ssh_run(ssh, ln_cmd)
                output = self.ssh_run(ssh, cmd, False, False)
            report_path = output.splitlines()[-1].split('：')[-1]
            self.log.info(f'获取到的元数据校验报告:{report_path}')
            report = self.ssh_run(ssh, f'cat {report_path}', False, False)

            for line in report.splitlines():
                if not line.startswith('**校验结果**: '):
                    continue
                res = line.split(': ')[1].strip()
                if res == '成功':
                    self.log.info(f'元数据校验结果：{res}')
                else:
                    self.log.info(f'元数据校验结果：{res}，详情请查看：{report_path}')
                return res, report_path

    def check_prometheus(self):
        """
        检查prometheus进程是否在运行，若未运行则重新拉起
        :return:无
        """
        path = os.path.join('/log', 'prometheus-2.53.0.linux-amd64')
        cmd = f'cd {path};' \
              f'nohup ./prometheus --config.file=prometheus.yml ' \
              f'> prometheus.log 2>&1 & '
        with self.ssh_connect(self.prometheus_ip, 'root', self.passwd) as ssh:
            check_res = self.ssh_run(ssh, 'ps -ef|grep prometheus', False,
                                     True)
            if 'prometheus.yml' not in check_res:
                self.ssh_run(ssh, cmd, False, True)
                self.log.info(f'prometheus已启动，'
                              f'访问{self.prometheus_ip}:9090/targets')
            else:
                self.log.info(f'prometheus运行中, '
                              f'访问{self.prometheus_ip}:9090/targets')

    def check_node_exporter(self):
        """
        检查node_export进程是否在运行，若未运行则重新拉起
        :return: 无
        """
        path = os.path.join('/log', 'node_exporter-1.8.1.linux-amd64')
        cmd = f'cd {path};' \
              f'nohup ./node_exporter > node_exporter.log 2>&1 & '
        for ip in self.ip_lists:
            self.log.info(ip)
            with self.ssh_connect(ip, 'root', self.passwd) as ssh:
                check_res = self.ssh_run(ssh, 'ps -ef|grep node_exporter',
                                         False, True)
                if './node_exporter' not in check_res:
                    self.ssh_run(ssh, cmd, False, True)
                    self.log.info(f'{ip}:node_exporter已启动')
                else:
                    self.log.info(f'{ip}:node_exporter运行中')

    def start_opengauss_exporter(self, **kwargs):
        """
        配置、启动opengauss-prometheus-exporter
        :param kwargs: 环境信息
        :return: 无
        """
        path = os.path.join('/log', 'opengauss-prometheus-exporter-master')
        exec_path = os.path.join(path, 'bin', 'opengauss_exporter')
        with self.ssh_connect(self.ip_lists[0], self.upgrade_user,
                              self.passwd) as ssh:
            cmd = f'''source {kwargs['db_env']}; 
            gs_guc reload -N all -I all -c "password_encryption_type=1" 
            gs_om -t restart
            '''
            self.ssh_run(ssh, cmd, False, True)
            sql = f'''create user {self.prometheus_user} with monadmin 
            password '{self.passwd}';
            grant usage on schema dbe_perf to {self.prometheus_user};
            grant select on pg_stat_replication to {self.prometheus_user};
            alter role {self.prometheus_user} sysadmin;
            '''
            self.ssh_run_sql(ssh, sql, **self.upgrade)

        for ip in self.ip_lists:
            self.log.info(ip)
            with self.ssh_connect(ip, 'root', self.passwd) as ssh:
                cmd = f'chown -R {self.upgrade_user}:{self.upgrade_user} ' \
                      f'{path}'
                self.ssh_run(ssh, cmd, False, True)
            with self.ssh_connect(ip, self.upgrade_user, self.passwd) as ssh:
                cmd = f'''source {kwargs['db_env']}; 
                gs_guc set -N all -I all -h "host all all {ip}/32 md5"; 
                gs_om -t restart
                '''
                self.ssh_run(ssh, cmd, False, True)
                env = f'export DATA_SOURCE_NAME="host={ip} ' \
                      f'user={self.prometheus_user} password={self.passwd} ' \
                      f'port={self.dbport} dbname={self.db_name} ' \
                      f'sslmode=disable"'
                cmd = f'''echo '{env}' >> {kwargs['db_env']}
                '''
                self.ssh_run(ssh, cmd, False, True)
                cmd = f'cd {path};' \
                      f'source {kwargs["db_env"]};' \
                      f'nohup {exec_path} ' \
                      f'--config="og_exporter_default.yaml"' \
                      f'>>./og_exporter.log 2>&1 & ' \
                      f'ps ux'
                self.log.info(cmd)
                self.ssh_run(ssh, cmd, False, True)
                check_res = self.ssh_run(ssh, 'ps -ef|grep opengauss_exporter',
                                         False, True)
                if 'opengauss_exporter' in check_res:
                    self.log.info(f'{ip}:opengauss-prometheus-exporter已启动')


if __name__ == '__main__':
    if len(sys.argv) < 17:
        raise Exception('请检查入参数量！')
    up_params = dict()
    up_params['userpasswd'] = sys.argv[1]
    up_params['upgrade_ftp_path'] = sys.argv[2]
    up_params['upgrade_user_prefix'] = sys.argv[3]
    up_params['prefixPort'] = sys.argv[4]
    up_params['db_root_path'] = sys.argv[5]
    up_params['pkg_root_path'] = sys.argv[6]
    up_params['ipaddress'] = sys.argv[7]
    up_params['ipaddress_std1'] = sys.argv[8]
    up_params['ipaddress_std2'] = sys.argv[9]
    up_params['tpcc_path'] = sys.argv[16]
    hostnames = (sys.argv[10], sys.argv[11], sys.argv[12])
